/*
 * Copyright (c) 2018-2024 北京华恒盛世科技有限公司
 * QUICKPOOL is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *     http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
*/

#include "jm.h"

static int status_task(ee_request_t, struct jobTask *, unsigned int);
static void collect_task_usage(struct jobTask *);
static int send_task_result(struct jobTask *);
static void delete_jobtask(struct jobTask *);

/**
 * @brief       fork a child to execute the job task.
 *
 * @param[in]   jobtask      #1: job task to be executed
 *
 * @retval      0            #1: succeed
 * @retval      != 0         #2: fork fail
 *
 * @note        create cgroup entry for the job task.
 *              For child, we need close log and all channel fds.
 */
int
fork_2start_jobtask(struct jobTask *jobtask)
{
    char strJobId[32];

    pid_t pid;
    pid = fork();
    if (0 > pid) {
        log_message(ERR, NOCLASS, "%s: fork() failed starting job %s: %m",
                    __func__, jobid_2string(jobtask->jobid, strJobId));

        return ERROR_FORK_FAIL;
    }

    if (0 == pid) { // child to execute job

        close_log();
        gf_child_chan_close(-1);
        finalize_tcl();

        char logfile[PATH_MAX+1];
        snprintf(logfile, PATH_MAX, "%s/jmchild.%s.log",
                 baseParams[BASE_LOGDIR].string_value, gptrMyHost->name);
        open_log(logfile, baseParams[BASE_LOG_MASK].string_value,
                 baseParams[BASE_JM_TIME_DEBUG].int_value);

        jobtask->task_pid = getpid();
        jobtask->task_pgid = jobtask->task_pid;

        if (setpgid(0, jobtask->task_pgid) < 0) {

            log_message(ERR, NOCLASS, STR_JOB_FAIL_S_M, __func__,
                        jobid_2string(jobtask->jobid, strJobId), "setpgid");

            jobtask->task_pgid = -1;
            exit(-1);
        }

        if (NULL != jobtask->execute_user && '\0' != jobtask->execute_user[0]) {
            gf_setenv("QP_JOB_EXECUSER", jobtask->execute_user);
        }

        if (NULL != jobtask->cwd && '\0' != jobtask->cwd[0]) {
            gf_setenv("QP_SUBCWD", jobtask->cwd);
        }

        char val[PATH_MAX];
        snprintf(val, 12, "%d", jobtask->node_rank);
        gf_setenv("QP_NODE_RANK", val);

        snprintf(val, 20, "%d", ARRAY_JOBID(jobtask->jobid));
        gf_setenv("QP_JOBID", val);
        snprintf(val, 12, "%d", ARRAY_INDEX(jobtask->jobid));
        gf_setenv ("QP_JOBINDEX", val);

        if (NULL != jobtask->script) {
            gf_setenv("QP_JOBFILENAME", jobtask->script);
        }

        struct passwd *pw;
        pw = get_user_byname(jobtask->execute_user);
        if (NULL == pw) {
            log_message(ERR, NOCLASS, "%s: No valid user <%s> found for job <%s> due to %m.",
                        __func__, jobtask->execute_user,
                        jobid_2string(jobtask->jobid, strJobId));
            exit(-1);
        }

        jobtask->execute_gid = pw->pw_gid;
        jobtask->execute_uid = pw->pw_uid;

        char tmpDirName[PATH_MAX+1];
        get_tmpdir_4job(tmpDirName, jobtask->jobid, jobtask->execute_uid);
        if ('\0' != tmpDirName[0]) {
            gf_setenv("TMPDIR", tmpDirName);
        }

        char strJobFileDir[PATH_MAX+1];
        if (pw->pw_dir && gf_isfullpath(pw->pw_dir)) {

            snprintf(strJobFileDir, PATH_MAX, "%s/.quickpool", pw->pw_dir);

            struct stat st;
            int ret;
            ret = stat(strJobFileDir, &st);
            if (0 > ret) {
                mkdir(strJobFileDir, 0700);
            }

            char strOutputFile[PATH_MAX+1], strErrFile[PATH_MAX+1];
            if (NULL != jobtask->script) {
                if (gf_isfullpath(jobtask->script)) {
                    snprintf(strOutputFile, PATH_MAX, "%s.%d.out",
                             jobtask->script, jobtask->node_rank);
                    snprintf(strErrFile, PATH_MAX, "%s.%d.err",
                             jobtask->script, jobtask->node_rank);
                } else {
                    snprintf(strOutputFile, PATH_MAX, "%s/%s.%d.out",
                             strJobFileDir, jobtask->script, jobtask->node_rank);
                    snprintf(strErrFile, PATH_MAX, "%s/%s.%d.err",
                            strJobFileDir, jobtask->script, jobtask->node_rank);
                }
            } else {
                snprintf(strOutputFile, PATH_MAX, "%s/%s.%d.out",
                         strJobFileDir, jobid_2longstring(jobtask->jobid, strJobId),
                         jobtask->node_rank);
                snprintf(strErrFile, PATH_MAX, "%s/%s.%d.err",
                         strJobFileDir, jobid_2longstring(jobtask->jobid, strJobId),
                         jobtask->node_rank);
            }

            int i;
            i = open(strOutputFile, O_WRONLY| O_CREAT| O_APPEND, 0600);
            if (0 > i) {
                log_message(ERR, NOCLASS, STR_JOB_FAIL_S_S_M, __func__,
                            jobid_2string(jobtask->jobid, strJobId), "open", strOutputFile);
                exit(-1);
            }

            if (dup2(i, 1) == -1) {
                log_message(ERR, NOCLASS, "%s: job <%s> task <%d> dup(%d,1) stdout <%s> failed due to %m.",
                            __func__, jobid_2longstring(jobtask->jobid, strJobId), i, strOutputFile);
                exit(-1);
            }

            i = open(strErrFile, O_WRONLY| O_CREAT| O_APPEND, 0600);
            if (0 > i) {
                log_message(ERR, NOCLASS, STR_JOB_FAIL_S_S_M, __func__,
                            jobid_2string(jobtask->jobid, strJobId), "open", strErrFile);
                exit(-1);
            }

            if (dup2(i, 2) == -1) {
                log_message(ERR, NOCLASS, "%s: job <%s> task <%d> dup(%d,1) stderr <%s> failed due to %m.",
                            __func__, jobid_2longstring(jobtask->jobid, strJobId), i, strErrFile);
                exit(-1);
            }
        }

        if (set_userid_4job(jobtask->jobid, jobtask->execute_user,
                            jobtask->execute_gid, jobtask->execute_uid) < 0) {
            exit(-1);
        }

        int i;
        for (i = 1; i < NSIG; i++) {
            Signal_(i, SIG_DFL);
        }

        Signal_(SIGHUP, SIG_IGN);

        sigset_t newmask;
        sigemptyset(&newmask);
        sigprocmask(SIG_SETMASK, &newmask, NULL);

        /* unblock all signals */
        alarm(0);

        char **execArgv;
        execArgv = set_startjob_arguments(NULL, jobtask->task_argv);
        if (NULL == execArgv) {
            exit(-1);
        }

        execvp(execArgv[0], execArgv);

        log_message(ERR, NOCLASS, "%s: unable to execute job task <%s> for job <%s> due to %m.",
                    __func__, jobtask->task_argv[0], jobid_2string(jobtask->jobid, strJobId));

        exit(99);
    }

    jobtask->task_pid = pid;
    jobtask->task_pgid = pid;
    jobtask->update_usage = FALSE;

    //set cgroup entry
    int rc;
    rc = cgroup_setup_job_entry(jobtask->jobid, jobtask->task_pid, "memory");
    if (rc < 0) {
        log_message(ERR, NOCLASS, "%s: failed to set cgroup(memory) for job[%s], pid[%d]",
                    __func__, jobid_2string(jobtask->jobid, strJobId), pid);
    }
    rc = cgroup_setup_job_entry(jobtask->jobid, jobtask->task_pid, "cpuacct");
    if (rc < 0) {
        log_message(ERR, NOCLASS, "%s: failed to set cgroup(cpuacct) for job[%s], pid[%d]",
                    __func__, jobid_2string(jobtask->jobid, strJobId), pid);
    }

    return 0;
} // end function fork_2start_jobtask

/**
 * @brief       Go through JM job task list and do check one by one.
 *
 * @note        check if job task is finished or not.
 *              collect job task usage and send to qptask.
 */
void
check_finish_tasks(void)
{
    struct jobTask *task, *next;

    for (task=(struct jobTask *)glistJobTask->forw; task!=(struct jobTask *)glistJobTask;
         task = next) {

        next = (struct jobTask *)task->forw;

        char strJobId[32];

        if (0 < task->end_time) {

            collect_task_usage(task);

            int ret;
            ret = status_task(BATCH_STATUS_TASK, task,
                              (task->start_time > gtimeNodeBoot) ? 0 : ERROR_HOST_REBOOT);
            if (0 > ret) {
                log_message(ERR, NOCLASS, "%s: failed to send job <%s> task starting to first node <%s>. status_task() return %d.",
                            __func__, jobid_2string(task->jobid, strJobId), task->from_node, ret);

                if (-2 != ret) {
                    task->comm_failcnt ++;
                    continue;
                }
            }

            // send out task stdout and stderr after task is finished.
            send_task_result(task);

            delete_jobtask(task);
        } else {

            int ret;
            ret = kill_task_pg(task, 0);
            if (0 > ret) {

                STATUS_WAIT_T wStatus;

                JOB_WSTATUS(wStatus) = 0xff00;
                task->w_status = JOB_WSTATUS(wStatus);
                task->end_time = time(NULL);

                ret = status_task(BATCH_STATUS_TASK, task,
                                  (task->start_time > gtimeNodeBoot) ? 0 : ERROR_HOST_REBOOT);

                if (-2 != ret) {
                    task->comm_failcnt ++;
                    continue;
                }

                delete_jobtask(task);
                continue;
            }

            if (!task->update_usage) {
                continue;
            }

            ret = status_task(BATCH_STATUS_TASK, task,
                              (task->start_time > gtimeNodeBoot) ? 0 : ERROR_HOST_REBOOT);
            if (0 > ret) {
                log_message(ERR, NOCLASS, "%s: failed to send job <%s> task starting to first node <%s>. status_task() return %d.",
                            __func__, jobid_2string(task->jobid, strJobId), task->from_node, ret);
                if (-2 != ret) {
                    task->comm_failcnt ++;
                    continue;
                }

                delete_jobtask(task);
            }
        }
    }

    return;
} // end function check_finish_tasks

/**
 * @brief       call kill to send signal to job task.
 *
 * @param[in]   task         #1: job task
 * @param[in]   sig          #2: signal to be sent
 *
 * @retval      0            #1: succeed
 * @retval      > 0          #2: kill return fail
 *
 * @note        if sig == 0, test if task pid has been gone.
 */
int
kill_task_pg(struct jobTask *task, int sig)
{
    char strJobId[32];
    log_message(DEBUG, DEBUGSIGNAL, "%s: Job <%s> task pid %d pgid %d sig %d",
                __func__, jobid_2string(task->jobid, strJobId), task->task_pid,
                task->task_pgid, sig);

    int pgid0[1], *pgid;
    if (0 == task->task_pgid) {
	if (kill(task->task_pid, sig) == 0) {
            kill(-task->task_pid, sig);
            return 0;
        }
        return (kill(-task->task_pid, sig));
    }

    if (0 == task->task_usage[0].npgids) {
        pgid = pgid0;
        pgid[0] = task->task_pgid;
    } else {
        pgid = task->task_usage[0].pgid;
    }

    collect_task_usage(task);

    unsigned int i;
    if (kill(task->task_pid, sig) == 0) {

        log_message(DEBUG, DEBUGSIGNAL, "%s: Job %s task kill(pid=%d) is good", 
                    __func__, jobid_2string(task->jobid, strJobId), task->task_pid);

        if (0 < task->task_usage[0].npgids) {
            for (i = 0; i < task->task_usage[0].npgids; i++) {
                kill(-task->task_usage[0].pgid[i], sig);
            }
        } else {
            kill(-task->task_pgid, sig);
        }

        return 0;
    } else {
        log_message(DEBUG, DEBUGSIGNAL, "%s: Job %s task kill(pid=%d) <%d> failed <%m>",
                    __func__, jobid_2string(task->jobid, strJobId),
                    task->task_pid, sig);
    }

    if (0 < task->task_usage[0].npgids) {
        unsigned int ndead = 0;

        for (i = 0; i < task->task_usage[0].npgids; i++) {
            if (kill(-task->task_usage[0].pgid[i], sig) < 0) {
               ndead++;

               log_message(DEBUG, DEBUGSIGNAL, "%s: Job %s task kill(%d) group are gone",
                            __func__, jobid_2string(task->jobid, strJobId),
                            task->task_usage[0].pgid[i]);
            }
        }

        if (ndead == task->task_usage[0].npgids) {
            log_message(DEBUG, DEBUGSIGNAL, "%s: Job %s task pgids all dead", 
                        __func__, jobid_2string(task->jobid, strJobId));
            return -1;
        }
    } else {
        if (kill(-task->task_pgid, sig) < 0) {
            log_message(DEBUG, DEBUGSIGNAL, "%s: Job %s task kill(%d) group is gone",
                        __func__, jobid_2string(task->jobid, strJobId),
                        task->task_pgid);
            return -1;
        }
    }

    log_message(DEBUG, DEBUGSIGNAL, "%s: Job %s task child gone, but some pgids still around",
                __func__, jobid_2string(task->jobid, strJobId));

    return 0;
} // end function kill_jobpg

void
read_job_acct(struct jobEntry *job)
{
    char tmpDirName[PATH_MAX+1];
    get_tmpdir_4job(tmpDirName, job->job_base.jobid, job->job_base.execute_uid);

    change_uid(job->job_base.execute_uid);

    char strAcctFile[PATH_MAX+1];
    FILE *fileAcct;
    unsigned int lineNum = 0;
    struct record *taskRec;

    unsigned int i;
    for (i=0; i<job->job_base.number_node; i++) {

        snprintf(strAcctFile, PATH_MAX, "%s/.%s.%d.acct",
                 tmpDirName, job->job_base.submit->script, i);

        fileAcct = fopen(strAcctFile, "r");
        if (NULL == fileAcct) {
            continue;
        } 

        lineNum = 0;
        taskRec = read_file_line(fileAcct, &lineNum);
        if (NULL == taskRec) {

            fclose(fileAcct);

            if (quickpool_errno != ERROR_EOF) {
                change_uid(guidManager);
                log_message(DEBUG, TRACER, "%s: failed to read job acct file <%s>.",
                            __func__, strAcctFile);
                change_uid(job->job_base.execute_uid);
                continue;
            }

            continue;
        }

        fclose(fileAcct);

        unsigned int j,numTask;
        unsigned int rank=0, pidIndex=0, pgidIndex=0;
        for (j= 0; j<taskRec->numKVs; j++) {
            switch (taskRec->KVs[j].key) {
            case RECORD_KEY_NUM_TASK:
                numTask = atoi(taskRec->KVs[j].value);
                if (numTask > job->job_base.number_node) {
                }
                break;
            case RECORD_KEY_NODE_RANK:
                rank = atoi(taskRec->KVs[j].value);
                gf_xdr_free(xdr_job_usage, &(job->inst_usage[rank]));
                break;
            case RECORD_KEY_JOB_PID:
                pidIndex = 0;
                pgidIndex = 0;
                break;
            case RECORD_KEY_TASK_STATUS:
                break;
            case RECORD_KEY_USAGE_MEM:
                job->inst_usage[rank].mem = atol(taskRec->KVs[j].value);
                break;
            case RECORD_KEY_USAGE_SWAP:
                job->inst_usage[rank].swap = atol(taskRec->KVs[j].value);
                break;
            case RECORD_KEY_USAGE_UTIME:
                job->inst_usage[rank].utime = atol(taskRec->KVs[j].value);
                break;
            case RECORD_KEY_USAGE_STIME:
                job->inst_usage[rank].stime = atol(taskRec->KVs[j].value);
                break;
            case RECORD_KEY_USAGE_NPID:
                job->inst_usage[rank].npids = atoi(taskRec->KVs[j].value);
                if (0 < job->inst_usage[rank].npids) {
                    struct pidInfo *info;
                    info = (struct pidInfo *)calloc(job->inst_usage[rank].npids+1,
                                                    sizeof(struct pidInfo));
                    if (NULL == info) {
                        return;
                    }
                    job->inst_usage[rank].pidinfo = info;
                }
                break;
            case RECORD_KEY_USAGE_PIDINFO:
                sscanf(taskRec->KVs[j].value, "%d %d %d",
                       &(job->inst_usage[rank].pidinfo[pidIndex].pid),
                       &(job->inst_usage[rank].pidinfo[pidIndex].ppid),
                       &(job->inst_usage[rank].pidinfo[pidIndex].pgid));

                pidIndex ++;
                break;
            case RECORD_KEY_USAGE_NPGID:
                job->inst_usage[rank].npgids = atoi(taskRec->KVs[j].value);
                if (0 < job->inst_usage[rank].npgids) {
                    job->inst_usage[rank].pgid = (int *)calloc(job->inst_usage[rank].npgids+1,
                                                                sizeof(int));
                    if (NULL == job->inst_usage[rank].pgid) {
                        return;
                    }
                }
                break;
            case RECORD_KEY_USAGE_PGID:
                job->inst_usage[rank].pgid[pgidIndex++] = atoi(taskRec->KVs[j].value);
                break;
            }
        }
    }

    snprintf(strAcctFile, PATH_MAX, "%s/.%s.acct", tmpDirName, job->job_base.submit->script);

    fileAcct = fopen(strAcctFile, "r");
    if (NULL == fileAcct) {
        change_uid(guidManager);
        return;    
    }

    lineNum = 0;
    taskRec = read_file_line(fileAcct, &lineNum);
    if (NULL == taskRec) {

        fclose(fileAcct);
        change_uid(guidManager);

        if (quickpool_errno != ERROR_EOF) {
            log_message(DEBUG, TRACER, "%s: failed to read job acct file <%s>.",
                        __func__, strAcctFile);
            return;
        }

        return;
    }

    fclose(fileAcct);

    unsigned int j,numTask;
    unsigned int rank=0, pidIndex=0, pgidIndex=0;
    for (j= 0; j<taskRec->numKVs; j++) {
        switch (taskRec->KVs[j].key) {
        case RECORD_KEY_NUM_TASK:
            numTask = atoi(taskRec->KVs[j].value);
            if (numTask > job->job_base.number_node) {
            }
            break;
        case RECORD_KEY_NODE_RANK:
            rank = atoi(taskRec->KVs[j].value);
            gf_xdr_free(xdr_job_usage, &(job->inst_usage[rank]));
            break;
        case RECORD_KEY_JOB_PID:
            pidIndex = 0;
            pgidIndex = 0;
            break;
        case RECORD_KEY_TASK_STATUS:
            break;
        case RECORD_KEY_USAGE_MEM:
            job->inst_usage[rank].mem = atol(taskRec->KVs[j].value);
            break;
        case RECORD_KEY_USAGE_SWAP:
            job->inst_usage[rank].swap = atol(taskRec->KVs[j].value);
            break;
        case RECORD_KEY_USAGE_UTIME:
            job->inst_usage[rank].utime = atol(taskRec->KVs[j].value);
            break;
        case RECORD_KEY_USAGE_STIME:
            job->inst_usage[rank].stime = atol(taskRec->KVs[j].value);
            break;
        case RECORD_KEY_USAGE_NPID:
            job->inst_usage[rank].npids = atoi(taskRec->KVs[j].value);
            if (0 < job->inst_usage[rank].npids) {
                struct pidInfo *info;
                info = (struct pidInfo *)calloc(job->inst_usage[rank].npids+1,
                                                sizeof(struct pidInfo));
                if (NULL == info) {
                    return;
                }
                job->inst_usage[rank].pidinfo = info;
            }
            break;
        case RECORD_KEY_USAGE_PIDINFO:
            sscanf(taskRec->KVs[j].value, "%d %d %d",
                   &(job->inst_usage[rank].pidinfo[pidIndex].pid),
                   &(job->inst_usage[rank].pidinfo[pidIndex].ppid),
                   &(job->inst_usage[rank].pidinfo[pidIndex].pgid));

            pidIndex ++;
            break;
        case RECORD_KEY_USAGE_NPGID:
            job->inst_usage[rank].npgids = atoi(taskRec->KVs[j].value);
            if (0 < job->inst_usage[rank].npgids) {
                job->inst_usage[rank].pgid = (int *)calloc(job->inst_usage[rank].npgids+1,
                                                           sizeof(int));
                if (NULL == job->inst_usage[rank].pgid) {
                    return;
                }
            }
            break;
        case RECORD_KEY_USAGE_PGID:
            job->inst_usage[rank].pgid[pgidIndex++] = atoi(taskRec->KVs[j].value);
            break;
        }
    }

    change_uid(guidManager);

    return;    
} // end function read_job_acct

/* ------------------------------------------------
 *
 *  below are static functions used in this file
 *
 * ------------------------------------------------ */

/**
 * @brief       Tell qptask that my running task status or task resource usage is changed.
 *
 * @param[in]   reqcode      #1: BATCH_STATUS_TASK
 * @param[in]   task         #2: my running job task
 * @param[in]   error        #4: if jm reboot or not
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed
 */
static int
status_task(ee_request_t reqcode, struct jobTask *task, unsigned int error)
{
    if (NULL == task) {
        return 0;
    }

    task->exit_code = error;

    struct request statusRequest;
    memset(&statusRequest, 0, sizeof(struct request));

    unsigned int num=4; // task_pid, task_pgid, node_rank, w_status

    if (0 < task->start_time) {
        num ++;
    }

    if (NULL != task->task_usage) {

        if (0 < task->task_usage->mem) {
            num ++;
        }

        if (0 < task->task_usage->swap) {
            num ++;
        }

        if (0 < task->task_usage->utime) {
            num ++;
        }

        if (0 < task->task_usage->stime) {
            num ++;
        }

        if (0 < task->task_usage->npids) {
            num ++;
            num += task->task_usage->npids;
        }

        if (0 < task->task_usage->npgids) {
            num ++;
            num += task->task_usage->npgids;
        }
    }

    if (0 < task->exit_code) {
        num ++;
    }

    if (0 < task->end_time) {
        num ++;
    }

    statusRequest.number = num;
    statusRequest.keyvalues = (struct keyvalue *)calloc(num, sizeof(struct keyvalue));
    if (NULL == statusRequest.keyvalues) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", statusRequest.number*sizeof(struct keyvalue));
        relaunch();
    }

    num = 0;

    statusRequest.keyvalues[num].key = REQUEST_KEY_NODE_RANK;
    statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == statusRequest.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(statusRequest.keyvalues[num].value, 12, "%d", task->node_rank);
    num ++;

    statusRequest.keyvalues[num].key = REQUEST_KEY_JOBPID;
    statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == statusRequest.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(statusRequest.keyvalues[num].value, 12, "%d", task->task_pid);
    num ++;

    statusRequest.keyvalues[num].key = REQUEST_KEY_JOBPGID;
    statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == statusRequest.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(statusRequest.keyvalues[num].value, 12, "%d", task->task_pgid);
    num ++;

    statusRequest.keyvalues[num].key = REQUEST_KEY_EXITSTATUS;
    statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == statusRequest.keyvalues[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
        relaunch();
    }
    snprintf(statusRequest.keyvalues[num].value, 12, "%d", task->w_status);
    num ++;

    if (0 < task->start_time) {
        statusRequest.keyvalues[num].key = REQUEST_KEY_START_TIME;
        statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == statusRequest.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(statusRequest.keyvalues[num].value, 12, "%ld", task->start_time);
        num ++;
    }

    if (NULL != task->task_usage) {
        if (0 < task->task_usage->mem) {
            statusRequest.keyvalues[num].key = REQUEST_KEY_USAGE_MEM;
            statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == statusRequest.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(statusRequest.keyvalues[num].value, 12, "%ld", task->task_usage->mem);
            num ++;
        }

        if (0 < task->task_usage->swap) {
            statusRequest.keyvalues[num].key = REQUEST_KEY_USAGE_SWAP;
            statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == statusRequest.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(statusRequest.keyvalues[num].value, 12, "%ld", task->task_usage->swap);
            num ++;
        }

        if (0 < task->task_usage->utime) {
            statusRequest.keyvalues[num].key = REQUEST_KEY_USAGE_UTIME;
            statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == statusRequest.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(statusRequest.keyvalues[num].value, 12, "%ld", task->task_usage->utime);
            num ++;
        }

        if (0 < task->task_usage->stime) {
            statusRequest.keyvalues[num].key = REQUEST_KEY_USAGE_STIME;
            statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == statusRequest.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(statusRequest.keyvalues[num].value, 12, "%ld", task->task_usage->stime);
            num ++;
        }

        unsigned int i;
        if (0 < task->task_usage->npids) {
            statusRequest.keyvalues[num].key = REQUEST_KEY_USAGE_NPID;
            statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == statusRequest.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(statusRequest.keyvalues[num].value, 12, "%d", task->task_usage->npids);
            num ++;

            for (i=0; i<task->task_usage->npids; i++) {
                statusRequest.keyvalues[num].key = REQUEST_KEY_USAGE_PIDINFO;
                statusRequest.keyvalues[num].value = (char *)calloc(36, sizeof(char));
                if (NULL == statusRequest.keyvalues[num].value) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 36);
                    relaunch();
                }

                snprintf(statusRequest.keyvalues[num].value, 36, "%d %d %d",
                         task->task_usage->pidinfo[i].pid, task->task_usage->pidinfo[i].ppid,
                         task->task_usage->pidinfo[i].pgid);
                num ++;
            }
        }

        if (0 < task->task_usage->npgids) {
            statusRequest.keyvalues[num].key = REQUEST_KEY_USAGE_NPGID;
            statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == statusRequest.keyvalues[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                relaunch();
            }
            snprintf(statusRequest.keyvalues[num].value, 12, "%d", task->task_usage->npgids);
            num ++;

            for (i=0; i<task->task_usage->npgids; i++) {
                statusRequest.keyvalues[num].key = REQUEST_KEY_USAGE_PGID;
                statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
                if (NULL == statusRequest.keyvalues[num].value) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
                    relaunch();
                }
                snprintf(statusRequest.keyvalues[num].value, 12, "%d", task->task_usage->pgid[i]);
                num ++;
            }
        }
    }

    if (0 < task->exit_code) {
        statusRequest.keyvalues[num].key = REQUEST_KEY_FAIL_REASON;
        statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == statusRequest.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(statusRequest.keyvalues[num].value, 12, "%d", task->exit_code);
        num ++;
    }

    if (0 < task->end_time) {
        statusRequest.keyvalues[num].key = REQUEST_KEY_END_TIME;
        statusRequest.keyvalues[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == statusRequest.keyvalues[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", 12);
            relaunch();
        }
        snprintf(statusRequest.keyvalues[num].value, 12, "%ld", task->end_time);
        num ++;
    }

    unsigned int size;
    size = gf_xdrsize_request(&statusRequest);
    size += HEADER_LEN;

    char *request;
    if ((request = (char *)calloc(size, sizeof(char))) == NULL) {
        gf_xdr_free(gf_xdr_request, &statusRequest);
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", size);
        relaunch();
    }

    struct header hdr;
    gf_init_header(&hdr);
    hdr.reqcode = reqcode;

    XDR xdrs;
    xdrmem_create(&xdrs, request, size, XDR_ENCODE);

    char strJobId[32];
    if (!gf_xdr_message(&xdrs, &statusRequest, &hdr, NULL, gf_xdr_request, NULL)) {
        log_message(ERR, NOCLASS, STR_JOB_FAIL_S, __func__,
                    jobid_2string(task->jobid, strJobId), "gf_xdr_message/gf_xdr_request");
        xdr_destroy(&xdrs);
        FREEUP(request);
        gf_xdr_free(gf_xdr_request, &statusRequest);
        return -1;
    }

    int cc;
    cc = gf_callslave(task->from_node, task->port, g_conntimeout, g_recvtimeout,
                      XDR_GETPOS(&xdrs), request, NULL, NULL, &(task->conn_fd));

    xdr_destroy(&xdrs);
    FREEUP(request);
    gf_xdr_free(gf_xdr_request, &statusRequest);

    if (0 > cc) {
        log_message(DEBUG, EXECUTE, "%s: gf_callslave(%s:%d) for job <%s> failed.",
                    __func__, task->from_node, task->port, jobid_2string(task->jobid, strJobId));
        return cc;
    }

    task->update_usage = FALSE;
    task->report_time = time(NULL);

    return 0;
} // end function status_task

static void
collect_task_usage(struct jobTask *task)
{
    struct jobResourceUsage *jru = NULL;
    RECORD_FUNCTIME(0, jru = cgroup_get_job_rusage(task->jobid), "cgroup_get_job_rusage");

    if (NULL == jru) {
        return;
    }

    char strJobId[32];

    unsigned int i;
    if (0 < jru->npgids) {

        static int jmPgid = -1;
        if (jmPgid < 0) {
            jmPgid =  getpgrp();
        }

        for (i=0; i<jru->npgids; i++) {

            if (jmPgid == jru->pgid[i]) {

                log_message(DEBUG, DEBUGSIGNAL, "%s: jm's pgid <%d> is the same as job <%s>",
                            __func__, jmPgid, jobid_2string(task->jobid, strJobId));

                unsigned int j;
                for (j=i; j < jru->npgids - 1; j++) {
                    jru->pgid[j] = jru->pgid[j+1];
                }
                jru->npgids--;
            }
        }

        if (0 == jru->npgids) {
            FREEUP(jru->pidinfo);
            FREEUP(jru->pgid);
            FREEUP(jru);
            return;
        }
    }

    time_t checkNow = time(NULL);
    if (checkNow - task->report_time > gintUpdateUsage * gintJMSleepTime) {

        log_message(DEBUG, DEBUGSIGNAL, "%s: will report job <%s> task usage to qptask soon.",
                    __func__, jobid_2string(task->jobid, strJobId));

        task->update_usage = TRUE;
    }

    copy_job_resourceusage(&(task->task_usage[0]), jru);

    if (0 < jru->npgids) {
       FREEUP(jru->pgid);
    }

    if (0 < jru->npids) {
        FREEUP(jru->pidinfo);
    }
    FREEUP(jru);

    return;
} // end function collect_task_usage

static int
send_task_result(struct jobTask *task)
{
    if (NULL == task) {
        return 0;
    }

    char strJobId[32];

    struct passwd *pw;
    pw = get_user_byname(task->execute_user);
    if (NULL == pw) {
        log_message(ERR, NOCLASS, "%s: No valid user <%s> found for job <%s> due to %m.",
                    __func__, task->execute_user,
                    jobid_2string(task->jobid, strJobId));
        return -1;
    }

    char strJobFileDir[PATH_MAX+1];
    if (pw->pw_dir && gf_isfullpath(pw->pw_dir)) {

        snprintf(strJobFileDir, PATH_MAX, "%s/.quickpool", pw->pw_dir);

        char strOutputFile[PATH_MAX+1], strErrFile[PATH_MAX+1];
        if (NULL != task->script) {
            if (gf_isfullpath(task->script)) {
                snprintf(strOutputFile, PATH_MAX, "%s.%d.out", task->script, task->node_rank);
                snprintf(strErrFile, PATH_MAX, "%s.%d.err", task->script, task->node_rank);
            } else {
                snprintf(strOutputFile, PATH_MAX, "%s/%s.%d.out",
                         strJobFileDir, task->script, task->node_rank);
                snprintf(strErrFile, PATH_MAX, "%s/%s.%d.err",
                         strJobFileDir, task->script, task->node_rank);
                }
        } else {
            snprintf(strOutputFile, PATH_MAX, "%s/%s.%d.out",
                     strJobFileDir, jobid_2longstring(task->jobid, strJobId), task->node_rank);
            snprintf(strErrFile, PATH_MAX, "%s/%s.%d.err",
                     strJobFileDir, jobid_2longstring(task->jobid, strJobId), task->node_rank);
        }

        unsigned int size;
        size = HEADER_LEN;

        struct stat st;
        int ret;
        ret = stat(strOutputFile, &st);
        if (0 == ret && 0 < st.st_size && 1024*1024*10 > st.st_size) {
            size += st.st_size;
        }

        ret = stat(strErrFile, &st);
        if (0 == ret && 0 < st.st_size && 1024*1024*10 > st.st_size) {
            size += st.st_size;
        }
    }

    return 0;
} // end function send_task_result

static void
delete_jobtask(struct jobTask *task)
{
    /* Clean up cgroup enry for job. */
    cgroup_rm_job_entry(task->jobid, "memory");
    cgroup_rm_job_entry(task->jobid, "cpuacct");

    gf_chan_close(task->conn_fd);

    struct passwd *pw;
    pw = get_user_byname(task->execute_user);
    if (NULL != pw) {

        char strJobFileDir[PATH_MAX+1];
        if (pw->pw_dir && gf_isfullpath(pw->pw_dir)) {

            snprintf(strJobFileDir, PATH_MAX, "%s/.quickpool", pw->pw_dir);

            char strOutputFile[PATH_MAX+1], strErrFile[PATH_MAX+1];
            if (NULL != task->script) {
                if (gf_isfullpath(task->script)) {
                    snprintf(strOutputFile, PATH_MAX, "%s.%d.out",
                             task->script, task->node_rank);
                    snprintf(strErrFile, PATH_MAX, "%s.%d.err",
                             task->script, task->node_rank);
                } else {
                    snprintf(strOutputFile, PATH_MAX, "%s/%s.%d.out",
                             strJobFileDir, task->script, task->node_rank);
                    snprintf(strErrFile, PATH_MAX, "%s/%s.%d.err",
                             strJobFileDir, task->script, task->node_rank);
                }
            } else {
                char strJobId[32];
                snprintf(strOutputFile, PATH_MAX, "%s/%s.%d.out",
                         strJobFileDir, jobid_2longstring(task->jobid, strJobId),
                         task->node_rank);
                snprintf(strErrFile, PATH_MAX, "%s/%s.%d.err",
                         strJobFileDir, jobid_2longstring(task->jobid, strJobId),
                         task->node_rank);
            }

            unlink(strOutputFile);
            unlink(strErrFile);
        }
    }

    char tmpDirName[PATH_MAX+1];
    get_tmpdir_4job(tmpDirName, task->jobid, task->execute_uid);

    char strAcctFile[PATH_MAX+1];
    snprintf(strAcctFile, PATH_MAX, "%s/%lld.ee.acct", tmpDirName, task->jobid);
    unlink(strAcctFile);

    sigset_t newmask, oldmask;
    sigemptyset(&newmask);
    sigaddset(&newmask, SIGCHLD);
    sigprocmask(SIG_BLOCK, &newmask, &oldmask);
    gf_list_remove(glistJobTask, (LIST_T *)task);
    sigprocmask(SIG_SETMASK, &oldmask, NULL);

    FREEUP(task->from_node);
    FREEUP(task->execute_user);
    FREEUP(task->cwd);
    FREEUP(task->script);

    if (NULL != task->task_usage) {
        FREEUP(task->task_usage->pidinfo);
        FREEUP(task->task_usage->pgid);
    }
    FREEUP(task->task_usage);

    int i;
    for (i=0; task->task_argv[i]; i++) {
        FREEUP(task->task_argv[i]);
    }
    FREEUP(task->task_argv);

    FREEUP(task);

    return;
} // end function delete_jobtask
